package vip.meeet.kafka.consumer;

import cn.freemethod.kafka.Configuration;
import cn.freemethod.kafka.listener.ReceiveListener;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.consumer.Whitelist;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Properties;


public class MessageConsumer implements Runnable {
	
    private static final String KAFKA_PROPERTIES = "kafka-consumer.properties";

	private static final Logger logger=LoggerFactory.getLogger(MessageConsumer.class);

    private ConsumerConnector javaConsumer;

    private  String threadName;
    
    private String cumsumerGroupId;
    
    private String topic;
    
    private ReceiveListener listener;


    MessageConsumer(String threadName, String cumsumerGroupId, ReceiveListener listener, String topic) {
        this.threadName=threadName;
        this.cumsumerGroupId=cumsumerGroupId;
        this.listener = listener;
        this.topic=topic;
    }

    private void connect() {
        logger.info("connect kafka......");
        if(javaConsumer != null) {
            javaConsumer.shutdown();
            javaConsumer=null;
        }
        Properties props=Configuration.getPropertiesByName(KAFKA_PROPERTIES);
        props.setProperty("group.id", cumsumerGroupId);
        ConsumerConfig consumerConfig = new ConsumerConfig(props);
        javaConsumer=kafka.consumer.Consumer.createJavaConsumerConnector(consumerConfig);
        logger.info("connect kafka success");
    }

    @Override
    public void run() {
        Thread.currentThread().setName(threadName);
        logger.info(String.format("%s start listener ......", threadName));
        connect();
        while(true) {
            try {

                Whitelist list=new Whitelist(topic);
                List<KafkaStream<byte[], byte[]>> partitions=javaConsumer.createMessageStreamsByFilter(list);
                if(!partitions.isEmpty()){
                    for(KafkaStream<byte[], byte[]> stream: partitions) {
                        for(MessageAndMetadata<byte[], byte[]> msgAndMetadata: stream) {
                            handleMsg(msgAndMetadata);
                        }
                    }
                }
            } catch(Exception e) {
                logger.error("kafka connect error >> " + e.getMessage(), e);
                connect();
            }

            try {
                Thread.sleep(5000);
            } catch(InterruptedException e) {
                logger.error(e.getMessage());
            }
        }
    }
    private void handleMsg(MessageAndMetadata<byte[], byte[]> msgAndMetadata) {
        try {
            String msg=new String(msgAndMetadata.message(), "utf-8");
            logger.info(String.format("receive message >> %s", msg));
            listener.onReceive(msg);
        } catch(Exception e) {
            logger.error("The processing results of failure. message >> " + e.getMessage(), e);
        }
    }

}